/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.amit.concurrency;

import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import sun.nio.cs.ext.ISCII91;

/**
 *
 * @author amit
 */
public class Manager {

    AtomicBoolean stopThreads = new AtomicBoolean(false);
    AtomicBoolean finallyDone = new AtomicBoolean(false);
    public ArrayList<Operator> operators;
    Thread[] tArr;
    Thread watchDog;
    Channel sink;
    CyclicBarrier barrier;

    public Manager(Channel sink, ArrayList<Operator> operators) {
        this.sink = sink;
        this.operators = operators;
        int threadCount = 0;
        for (Operator operator : operators) {
            threadCount += operator.numThreads;
        }
        tArr = new Thread[threadCount];
        barrier = new CyclicBarrier(threadCount, new Runnable() {

            public void run() {
                if (checkDone()) {
                    finallyDone.set(true);
                }
            }
        });
    }

    public void stopThreads() {
        stopThreads.set(true);
        while (!finallyDone.get()) {
        }
        for (int i = 0; i < tArr.length; i++) {
            try {
                System.out.println("done with " + i);
                tArr[i].join();
            } catch (InterruptedException ex) {
                Logger.getLogger(Manager.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    /**
     * Checks to see if all of the work has been done.
     * This is checked by ensuring that the only queue with any items left is the sinks output channel.
     * Of course the sinks output channel can also be empty, indicating that all channels are empty.
     * @return true if done and false if not
     */
    public boolean checkDone() {
        int emptyChannels = 0;
        for (Operator operator : operators) {
            if (operator.channel != sink && (operator.channel.input.size() > 0 || operator.channel.output.size() > 0)) {
                return false;
            }

        }
        if (sink.input.size() == 0) {
            return true;
        }

        return false;
    }

    public void startThreads() {
        int i = 0;
        for (final Operator operator : operators) {
            Runnable r = new Runnable() {

                public void run() {
                    while (!finallyDone.get()) {
                        operator.runNext();
                        if (stopThreads.get()) {
                            try {
                                barrier.await();
                            } catch (Exception ex) {
                                Logger.getLogger(Manager.class.getName()).log(Level.SEVERE, null, ex);
                            }
                        }

                    }

                }
            };
            for (int j = 0; j <
                    operator.numThreads; j++) {
                tArr[i] = new Thread(r);
                tArr[i].start();
                i++;

            }


        }
    }

    public String getStatus() {
        StringBuffer buf = new StringBuffer();
        buf.append("-----------" + new Date() + "-------------\n");
        for (Operator op : operators) {
            buf.append(op.channel.channelName + ": IN: " + op.channel.input.size() + " OUT: " + op.channel.output.size() + "\n");
            buf.append("-----------------------------------\n");
        }

        return buf.toString();
    }
}
